Branching — Conditional Task Execution

BranchPythonOperator

The BranchPythonOperator lets you choose which downstream task to execute based on a condition:

from airflow.providers.standard.operators.python import BranchPythonOperator

def decide_branch(**kwargs):
    """Return the task_id of the branch to follow."""
    hour = kwargs['execution_date'].hour

    if hour < 12:
        return 'morning_pipeline'
    else:
        return 'evening_pipeline'

branch = BranchPythonOperator(
    task_id="choose_pipeline",
    python_callable=decide_branch,
)

morning = PythonOperator(task_id="morning_pipeline", ...)
evening = PythonOperator(task_id="evening_pipeline", ...)
join = EmptyOperator(task_id="join", trigger_rule="none_failed_min_one_success")

branch >> [morning, evening] >> join

graph TD
    A["choose_pipeline<br/>(BranchPythonOperator)"] -->|"if hour < 12"| B["morning_pipeline"]
    A -->|"if hour >= 12"| C["evening_pipeline"]
    B --> D["join<br/>(trigger_rule: none_failed_min_one_success)"]
    C --> D
    style A fill:#FF9800,stroke:#F57C00,color:#fff
    style B fill:#4CAF50,stroke:#388E3C,color:#fff
    style C fill:#2196F3,stroke:#1976D2,color:#fff
    style D fill:#607D8B,stroke:#455A64,color:#fff

TaskFlow Branching

@task.branch()
def decide_pipeline(data_size: int) -> str:
    if data_size > 1_000_000:
        return "run_spark_job"      # Big data → Spark
    elif data_size > 10_000:
        return "run_pandas_job"     # Medium → Pandas
    else:
        return "run_sql_query"      # Small → SQL

Trigger Rules

This content is reserved for Premium Members.